package spring.cloud.tasks.api.analyzer;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpaceConfiguration;
import spring.cloud.tasks.common.node.ExecutorNodePath;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ExecutorInfoAnalyzer {

    private static final Logger log = LoggerFactory.getLogger(ExecutorInfoAnalyzer.class);

    private final Map<String/** {executorName}-{domain} */, ExecutorStatistics> executorKeyToExecutorStatisticsMap = new ConcurrentHashMap<>();


    public void analyzeExecutor(ZooKeeper zooKeeper, NameSpaceConfiguration nameSpaceConfiguration) {
        // 该域的在线executor数量
        // 统计物理容器资源，统计版本数据
        if (!zooKeeper.checkExists(ExecutorNodePath.getExecutorsNodePath())) {
            return;
        }
        List<String> executorIdList = zooKeeper.getChildren(ExecutorNodePath.getExecutorsNodePath());
        if (executorIdList == null) {
            return;
        }
        for (String executorId : executorIdList) {
            // 在线的才统计
            if (zooKeeper.checkExists(ExecutorNodePath.getExecutorIpNodePath(executorId))) {
                // 统计是物理机还是容器
                String executorMapKey = nameSpaceConfiguration.getNameSpace() + "-" + executorId;
                ExecutorStatistics executorStatistics = executorKeyToExecutorStatisticsMap.get(executorMapKey);
                if (executorStatistics == null) {
                    executorStatistics = new ExecutorStatistics(nameSpaceConfiguration.getNameSpace(), executorId);
                    executorStatistics.setIp(zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(executorId)));
                    executorKeyToExecutorStatisticsMap.put(executorMapKey, executorStatistics);
                }
                executorStatistics.setLoadLevel(9999);//表示待定
                try {

                } catch (Exception e) {

                }
            }

        }

    }


    public void analyzeTaskExecutor(ZooKeeper zooKeeper, NameSpaceConfiguration nameSpaceConfiguration, List<String> executorIdList, String taskId, int loadLevel, TaskStatistics taskStatistics) {
        for (String executorId : executorIdList) {
            // 如果结点存活，算两样东西：1.遍历所有servers节点里面的processSuccessCount &
            // processFailureCount，用以统计任务每天的执行次数；2.统计executor的loadLevel;，
            if (!zooKeeper.checkExists(TaskNodePath.getExecutorStatus(taskId, executorId))) {
                continue;
            }


            // 2.统计executor的loadLevel
            calcTaskExecutorShardsAndLoadLevel(nameSpaceConfiguration.getNameSpace(), zooKeeper, taskId, loadLevel, taskStatistics, executorId);
        }
    }


    private void calcTaskExecutorShardsAndLoadLevel(String nameSpace, ZooKeeper zooKeeper, String job, int loadLevel, TaskStatistics taskStatistics, String executorId) {
        try {
            // enabled 的任务才需要计算权重
            if (!Boolean.parseBoolean(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(job, "enabled")))) {
                return;
            }

            String sharding = zooKeeper.getData(TaskNodePath.getServerSharding(job, executorId));
            if (StringUtils.isNotEmpty(sharding)) {
                // 更新job的executorsAndshards
                String exesAndShards = (taskStatistics.getExecutorsAndShards() == null ? "" : taskStatistics.getExecutorsAndShards()) + executorId + ":" + sharding + "; ";
                taskStatistics.setExecutorsAndShards(exesAndShards);
                // 2.统计是物理机还是容器
                String executorMapKey = nameSpace + "-" + executorId;
                ExecutorStatistics executorStatistics = executorKeyToExecutorStatisticsMap.get(executorMapKey);
                if (executorStatistics == null) {
                    executorStatistics = new ExecutorStatistics(nameSpace, executorId);
                    executorStatistics.setIp(zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(executorId)));
                    executorKeyToExecutorStatisticsMap.put(executorMapKey, executorStatistics);

                }
                if (executorStatistics.getJobAndShardings() != null) {
                    executorStatistics.setJobAndShardings(executorStatistics.getJobAndShardings() + job + ":" + sharding + ";");
                } else {
                    executorStatistics.setJobAndShardings(job + ":" + sharding + ";");
                }
                int newload = executorStatistics.getLoadLevel() + (loadLevel * sharding.split(",").length);
                executorStatistics.setLoadLevel(newload);
            }
        } catch (Exception e) {
            log.info(e.getMessage(), e);
        }
    }

    public List<ExecutorStatistics> getExecutorList() {
        return new ArrayList<>(executorKeyToExecutorStatisticsMap.values());
    }


}
